package com.pxene.dmp.mapper;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.pxene.dmp.common.URLUtils;
import com.pxene.dmp.constant.PatternList;
import com.pxene.dmp.constant.PatternList.Pattern;

public class GetWeatherCityMapper extends Mapper<LongWritable, Text, Text, Text> {

	private URL urlParser;
	private Text resultKey = new Text();
	private Text resultVal = new Text();

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
	}

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String[] columns = value.toString().split("\\|");
		if (columns.length != 7) {
			return;
		}
		String imsi = columns[0];
		String bsid = columns[1];
		String starttime = columns[2];
		String domain = columns[3];
		String url = columns[4];
		String province = columns[5];
		String dataType = columns[6];

		try {
			urlParser = new URL(url);
		} catch (MalformedURLException e) {
			return;
		}

		String city = "";

		PatternList weatherCityList = PatternList.load("weathercity.json");
		for (Pattern weatherCity : weatherCityList.elements()) {
			if (weatherCity.getCode().equals(domain)) {
				String expr = weatherCity.getParams()[0].getExpr();
				city = URLUtils.regexpExtract(url, expr).trim();
				break;
			}
		}

		if (city.length() > 0) {
			resultKey.set(imsi + "|" + bsid + "|" + starttime + "|" + domain + "|" + province + "|" + dataType);
			resultVal.set(city);
			context.write(resultKey, resultVal);
		}

	}
}
